package client;

import gui.ClientFrame;

import java.net.MalformedURLException;
import java.net.SocketException;
import java.net.URI;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;

import broker.Broker;
import broker.BrokerImpl.SubscriptionType;
import broker_manager.BrokerManager;

import common.BrokerRef;
import common.DefaultRMISecurityManager;
import common.NetworkParameters;
import common.PublishInfo;
import common.SubscriptionInfo;

public class ClientImpl extends UnicastRemoteObject implements Client {

	private class MultipublishThread extends Thread implements Runnable {

		private String message;
		private int period;
		private int step;
		private boolean stop = false;
		private String topic;

		public MultipublishThread(String message, String topic, int period,
				int step) {
			this.message = message;
			this.topic = topic;
			this.period = period;
			this.step = step;
			this.setName(topic);
		}

		@Override
		public void run() {
			try {
				for (int time = 0, count = 0; time < period; time += step, count++) {
					publish(message + " " + count, topic);
					Thread.sleep(step);
					if (stopMultipublishThreads || stop)
						return;
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

		// Thread-safely stop this thread
		public void stopThread() {
			stop = true;
		}
	}

	private static final long serialVersionUID = 1L;
	private BrokerManager brokerManager;
	private ClientFrame frame = null;
	private URI localURI;
	private ArrayList<MultipublishThread> multipublishThreads = new ArrayList<MultipublishThread>();
	private BrokerRef myBroker;
	private boolean running = false;
	private boolean stopMultipublishThreads = false;
	private ArrayList<String> subscribedTopics = new ArrayList<String>();

	private String username;

	public ClientImpl() throws RemoteException {
		super();
		if (System.getSecurityManager() == null) {
			System.setSecurityManager(new DefaultRMISecurityManager());
		}
	}

	public ClientImpl(ClientFrame frame) throws RemoteException {
		super();
		if (System.getSecurityManager() == null) {
			System.setSecurityManager(new DefaultRMISecurityManager());
		}
		this.frame = frame;
	}

	private void connectToNearestBroker() throws MalformedURLException,
			NotBoundException, RemoteException, Exception {

		// Get brokers list from broker manager
		ArrayList<URI> brokersURIs = brokerManager.getBrokers();
		while (brokersURIs.size() == 0) {
			System.out.println("No active brokers found, retrying in "
					+ NetworkParameters.backoffPeriod + " seconds");
			try {
				Thread.sleep(NetworkParameters.backoffPeriod * 1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			brokersURIs = brokerManager.getBrokers();
		}

		// Find minimum latency broker
		long lat[] = new long[NetworkParameters.numPings];
		long start, end, sum, minLatency = Long.MAX_VALUE;
		myBroker = null;
		Broker broker;
		System.out.println("Finding nearest broker");
		for (URI uri : brokersURIs) {
			System.out.print("Evaluating " + localURI + " --> " + uri + " -");
			try {
				broker = (Broker) Naming.lookup("//" + uri.toString()
						+ "/Broker");
				sum = 0;
				if (!uri.equals(localURI)) {
					for (int i = 0; i < lat.length; i++) {
						start = System.currentTimeMillis();
						broker.ping();
						end = System.currentTimeMillis();
						lat[i] = end - start;
						System.out.print(" " + lat[i] + "ms");
						sum += lat[i];
					}
				}
				System.out.println(" (ave: " + sum / lat.length + "ms)");
				if (sum < minLatency) {
					minLatency = sum;
					myBroker = new BrokerRef(uri, broker);
				}
			} catch (RemoteException e) {
				System.out
						.println("Link may be broken, latency set to infinity");
			}
		}
		System.out.println("Nearest broker: " + myBroker.uri);

		// Registering to nearest broker
		myBroker.broker.addClient(localURI);
	}

	public URI getLocalURI() {
		return localURI;
	}

	public BrokerRef getMyBroker() {
		return myBroker;
	}

	public ArrayList<String> getSubscribedTopics() {
		return subscribedTopics;
	}

	public void multipublish(String msg, String topic, int period, int step) {
		String message = "Message";
		if (msg != null && !msg.equals("")) {
			message = msg;
		}

		// Use a separate thread to prevent the sleeps in the iteration from
		// blocking the main thread.
		MultipublishThread thr = new MultipublishThread(message, topic, period,
				step);
		thr.start();
		multipublishThreads.add(thr);
	}

	@Override
	public void notifyBrokerShutdown() throws MalformedURLException,
			RemoteException, NotBoundException, Exception {

		// Update the progress bar
		frame.brokerShutdownProgress(0);

		// Find new nearest broker
		System.out.println("Nearest broker shutting down, finding another one");
		connectToNearestBroker();
		frame.brokerShutdownProgress(33);

		// Force previous remote subscriptions in each broker subscription
		// table, bypassing the queues
		for (URI brokerURI : brokerManager.getBrokers()) {
			Broker broker = (Broker) Naming.lookup("//" + brokerURI.toString()
					+ "/Broker");
			for (String topic : subscribedTopics) {
				broker.updateSubscriptionsTable(new SubscriptionInfo(
						myBroker.uri, myBroker.uri, topic,
						SubscriptionInfo.Action.ADD), SubscriptionType.REMOTE,
						false);
			}
		}
		frame.brokerShutdownProgress(66);

		// Force previous local subscriptions in nearest broker subscription
		// table, bypassing the queues
		for (String topic : subscribedTopics) {
			SubscriptionInfo info = new SubscriptionInfo(localURI, localURI,
					topic, SubscriptionInfo.Action.ADD);
			myBroker.broker.updateSubscriptionsTable(info,
					SubscriptionType.LOCAL, true);
		}
		frame.brokerShutdownProgress(100);
	}

	@Override
	public void printMessage(String sourceUsername, String msg, String topic) {
		System.out.println(sourceUsername + ": " + msg + " [" + topic + "]");
		if (frame != null)
			frame.printGuiMessage(sourceUsername, msg, topic);
	}

	public void publish(String msg, String topic) throws RemoteException {
		if (subscribedTopics.contains(topic)) {
			myBroker.broker.publish(new PublishInfo(msg, myBroker.uri, topic,
					null, username));
			System.out.println("Published [" + topic + ": " + msg + "]");
		} else {
			System.out.println("To publish in " + topic
					+ " you first need to subscribe to it.");
		}
	}

	public void start(String networkInterface, URI brokerManagerURI,
			String username) throws SocketException, MalformedURLException,
			RemoteException, NotBoundException, Exception {
		localURI = NetworkParameters.getLocalURI(networkInterface);
		if (username == null)
			this.username = localURI.toString();
		else
			this.username = username;
		Naming.rebind("Client", this);
		running = true;
		stopMultipublishThreads = false;
		System.out.println("Client started on " + localURI);
		brokerManager = (BrokerManager) Naming.lookup("//"
				+ brokerManagerURI.toString() + "/BrokerManager");
		System.out.println("Connected to broker manager "
				+ brokerManagerURI.toString());
		connectToNearestBroker();
	}

	public void stop(boolean stopRuntime) throws RemoteException, Exception,
			MalformedURLException, NotBoundException {
		System.out.println("Shutting down...");

		if (running) {

			// Set this flag to true to signal multipublish threads to stop
			stopMultipublishThreads = true;

			// Check if all the threads are actually stopped
			boolean allStopped;
			do {
				allStopped = true;
				for (Thread t : multipublishThreads) {
					if (t.isAlive()) {
						allStopped = false;
						System.out
								.println("Waiting for all multipublish threads to stop: retry in "
										+ NetworkParameters.backoffPeriod
										* 1000 + "sec");
						Thread.sleep(NetworkParameters.backoffPeriod * 1000);
						break;
					}
				}
			} while (!allStopped);

			myBroker.broker.removeClient(localURI);
			subscribedTopics.clear();
			Naming.unbind("Client");
			running = false;
		}
		if (stopRuntime) {
			System.out.println("Stopping runtime...");
			Runtime.getRuntime().exit(0);
		}
	}

	public boolean subscribe(String topic) throws RemoteException {
		System.out.println("Subscribing to topic " + topic);
		if (subscribedTopics.contains(topic)) {
			System.out.println("Already subscribed to that topic");
			return false;
		} else {
			subscribedTopics.add(topic);
			myBroker.broker.subscribe(new SubscriptionInfo(localURI, localURI,
					topic, SubscriptionInfo.Action.ADD));
			System.out.println("Subscribed to topic " + topic);
			return true;
		}
	}

	public void unsubscribe(String topic) throws RemoteException, Exception {

		// We need to stop the multipublish threads publishing on the topic i'm
		// unsubscribing
		for (MultipublishThread t : multipublishThreads) {
			if (t.getName().equals(topic))
				t.stopThread();
		}

		// Check if all the threads are actually stopped
		boolean allStopped;
		do {
			allStopped = true;
			for (MultipublishThread t : multipublishThreads) {
				if (t.getName().equals(topic) && t.isAlive()) {
					allStopped = false;
					System.out.println("Waiting for topic " + topic
							+ " multipublish threads to stop: retry in "
							+ NetworkParameters.backoffPeriod * 1000 + "sec");
					Thread.sleep(NetworkParameters.backoffPeriod * 1000);
					break;
				}
			}
		} while (!allStopped);

		System.out.println("Unsubscribing from topic " + topic);
		subscribedTopics.remove(topic);
		myBroker.broker.subscribe(new SubscriptionInfo(localURI, localURI,
				topic, SubscriptionInfo.Action.REMOVE));
		System.out.println("Unsubscribed to topic " + topic);
	}
}
